package com.world.rickcloudendpoint.stream;

import com.world.rickcloudendpoint.metrics.Instrumentation;
import lombok.extern.log4j.Log4j2;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Consumer;

/**
 * @description: 并行式任务
 * @author: jlsong
 * @date: 2022/02/28 22:54:23
 * @version: 1.0
 */
@Log4j2
public class Task {

    private final ExecutorService executorService;
    private int parallelism;
    private int threadCleanupDelay;
    private Consumer<Runnable> task;
    private Runnable taskFinishCallback;
    private final CountDownLatch countDownLatch;
    private final List<Future<?>> fnFutures;
    private Instrumentation instrumentation;

    /**
     * 实例化一个新任务
     *
     * @param parallelism        并发量
     * @param threadCleanupDelay 线程清理延迟
     * @param instrumentation    工具仪表
     * @param task               消费者任务
     */
    public Task(int parallelism, int threadCleanupDelay, Instrumentation instrumentation, Consumer<Runnable> task) {
        // 初始化线程池
        this.executorService = Executors.newFixedThreadPool(parallelism);
        this.parallelism = parallelism;
        this.threadCleanupDelay = threadCleanupDelay;
        this.task = task;
        // 初始化任务执行基数
        this.countDownLatch = new CountDownLatch(parallelism);
        // 任务返回结果
        this.fnFutures = new ArrayList<>(parallelism);
        this.taskFinishCallback = countDownLatch::countDown;
        this.instrumentation = instrumentation;
    }

    public Task run() {
        // 任务中创建多个线程，取等待消费者回调
        for (int i = 0; i < parallelism; i++) {
            fnFutures.add(executorService.submit(() -> {
                task.accept(taskFinishCallback);
            }));
        }
        return this;
    }

    /**
     * 等待完成
     *
     * @throws InterruptedException
     */
    public void waitForCompletion() throws InterruptedException {
        log.info("waiting for completion.");
        countDownLatch.await();
        ;
    }

    public Task stop() {
        try {
            log.info("Stopping task thread");
            fnFutures.forEach(consumerThread -> consumerThread.cancel(true));
            log.info("Sleeping thread during clean up for {} duration", threadCleanupDelay);
            Thread.sleep(threadCleanupDelay);
        } catch (InterruptedException e) {
            log.info("error stopping tasks,the reason is {} ", e.getMessage(), e);
        }
        return this;
    }
}
